package com.lianjia.sh.se.kafka;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by yangsp1 on 16/7/27.
 */
public class KafkaProducerTest {

    static KafkaProducer producer;


    public static void init(){
        Properties props = new Properties();
        //props.put("bootstrap.servers", "10.8.1.93:19092");
        //必须：kafka集群ip配置
        //props.put("bootstrap.servers", "10.8.1.160:19092,10.8.1.93:19092");
        //props.put("bootstrap.servers", "192.168.3.111:9092");
        //props.put("bootstrap.servers","115.28.23.129:19092");
        props.put("bootstrap.servers","192.168.3.59:9092,192.168.3.63:9092,192.168.3.64:9092");
        //props.put("bootstrap.servers","10.22.1.213:19092,10.22.1.213:19093");
        //props.put("auto.create.topics.enable",true);服务器端
        //消息逻辑id,即具有业务含义的唯一值，用于跟踪消息源（非必须）
        props.put("client.id","房源");
        //是否事务化（同步时吧）
        props.put("acks", "1");
        //是否失败重试：0不重试，大于0会重试
        props.put("retries", 0);
        //批处理消息大小（单位：字节）：缓冲超出此值就把消息发送到服务器
        props.put("batch.size", 64768 );
        //批处理时间间隔（单位：毫秒）：缓冲时间超过此值就把消息发送到服务器
        props.put("linger.ms", 0);

        //超时为1分钟
        props.put("request.timeout.ms",60000);
        //缓冲区大小：同一个生产者可以并发操作，因此可以缓冲多个线程数据
        props.put("buffer.memory", 1024 * 1024 * 5);


        //props.put("block.on.buffer.full",true);
        //key序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //value序列化器
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
    }


    public static void main(String args[]) throws Exception {
        Integer count = 1000000;//Integer.valueOf(args[0]);

        List<Exception> errors = new LinkedList<>();
        Map<String,Exception>  errors2 = new HashMap<>();
        String topic = "my-topic4";
        if(args.length > 1 && args[1] != null){
            topic = args[1];
        }
        init();
        Long start = System.currentTimeMillis();
        Integer rowCnt = count;
        String content = "第一条：＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝";
        producer.send(new ProducerRecord<String, String>(topic, "老杨叔叔", content));
        System.out.println(content);
        for(int i=0; i < rowCnt; i++) {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time = simpleDateFormat.format(new Date());
            content = "[" + i + "]用心感受大自然的力量,发送时间［" + time + "］";
            //System.out.println(content);
            ProducerRecord target = new ProducerRecord<String, String>(topic, "老杨叔叔", content);

                                                                      ;
            producer.send(target,new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception != null) {
                        errors.add(exception);
                        errors2.put(exception.getMessage(),exception);
                    }
                }
            });
            if(i % 10000 == 0 && i > 0) {
               // producer.flush();
            }

        }
        content = "最一条：＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝";
        producer.send(new ProducerRecord<String, String>(topic, "老杨叔叔", content));
        //producer.flush();,

        Long end = System.currentTimeMillis();
        System.out.println(content);
        System.out.println("发送 " + rowCnt + " 条数据用时：" + (end -start) + "毫秒");

        //Thread.sleep(1000*60*2);
        producer.flush();
        System.out.println("================结果=============");
        if(!errors.isEmpty()){
            System.out.println("发送失败数据量：" + errors.size());
        }
        if(!errors2.isEmpty()){
            for(String key : errors2.keySet()){
                System.out.println("发送失败数据：key=" + key +" ===== ");
            }
        }
        //producer.close();
        waitChildThreads();
    }

    public static void waitChildThreads() throws InterruptedException {
        int count = Thread.currentThread().getThreadGroup().activeCount();
        Thread[] list = new Thread[count];
        Thread.currentThread().getThreadGroup().enumerate(list);

        for(Thread thread : list){
            thread.join();
        }
    }
}